package com.holly.unit.kafka;

import com.holly.unit.kafka.exception.KafkaException;
import com.holly.unit.kafka.model.CallbackReceiveMessage;
import com.holly.unit.kafka.model.Kmessage;
import com.holly.unit.kafka.model.Ktopic;

import java.util.List;

/**
 * kafka 封装接口，每个接口都有demo说明
 * @author pengpeng
 * 
 * 配置如下：
 * holly: 
    kafka: 
      bootstrapServers: 172.26.166.57:19092,172.26.166.57:29092,172.26.166.57:39092
      ack: all
      
      #### 下面的序列化是json数据用的，如果传String数据，不用配置下面的
      keySerializer: org.apache.kafka.common.serialization.StringSerializer
      valueSerializer: com.holly.unit.kafka.serializer.JSONSerializer
      keyDeserializer: org.apache.kafka.common.serialization.StringDeserializer
      valueDeserializer: com.holly.unit.kafka.serializer.JSONDeserializer
      
      #### 其他参数请查看类 com.holly.unit.kafka.config.KafkaConfig
 */
public interface KafkaApi {
	/**
	 * 发送消息，错误抛出异常  保证
	 * @param kmessageList  发送消息的集合
	 * @throws RuntimeException
	 * DEMO:
	 *   @Autowired
	     private KafkaApi api;
		 try {
			 List<Kmessage> list = new  ArrayList<Kmessage>();
			 for(int i = 0;i<100000;i++)
			 {
				 list.add(new Kmessage(i+"", i+"", "g2", 0));
			 }
			api.sendMessage(list);
		} catch (Exception e) {
			System.out.println(e);
		}
	 */
	public void sendMessage(List<Kmessage> kmessageList) throws KafkaException;
	
	/**
	 * @param topic      订阅主题
	 * @param partition  订阅分区
	 * @param groupId    订阅分组ID
	 * @param callback   回调获取消息列表
	 * @param isAutoCommit  是否手动提交偏移量，防止数据丢失
	 * @return 消息集合
	 * @throws KafkaReceiveMessageException
	 * @param
	 * 	 @Autowired
	     private KafkaApi api;
		new Thread(()->{
			api.receiveMessage("g2", null, "worker",(List<Kmessage> messageList, Object kafkaConsumer,AtomicBoolean  isNoStop)->{
				
				-----------------逻辑处理--------------
				System.out.println(messageList);
				----------------逻辑处理结束---------------
				//如果isAutoCommit=false,则需要手动提交偏移量
                //KafkaConsumer kc = (KafkaConsumer) kafkaConsumer;
                //kc.commitSync();
				//如果想停止接收数据
				//isNoStop.set(false);
			}, true);	
		}).start();
	 */
	public void receiveMessage(String topic,Integer partition,String groupId,CallbackReceiveMessage callback,boolean isAutoCommit);
	/**
	 * 事务发送消息，保证精确一次投递，但是影响效率
	 * @param kmessageList    返回消息
	 * @throws RuntimeException
	 */
	public void sendMessageWithTranactions(List<Kmessage> kmessageList);
	

	/**
	 * 事务安全，保证数据顺序一致性，高可用。拉取数据不保证幂等性，需要去重
	 * 应用例子 比如获取订单topic，修改订单状态，然后，发到库存topic
	 * 数据转换流 从Topic "from" 接收数据并逻辑处理数据，再发送到 Topic "to"
	 * @param groupId    消费者分配组
	 * @param tranactionId  事务id,如果为null，则为默认事务id
	 * @param fromTopic {"topic":"xx","partition":null}  partition可以为bull
	 * @param toTopic   {"topic":"xx","partition":null}  partition可以为bull
	 * @param transformationCallBack
	 * 例子:
	 * 
	 * 		new Thread(()->{
			try{
				Ktopic fromTopic = new Ktopic("from", null);
				Ktopic toTopic = new Ktopic("to", null);

				kafkaApi.topicTransformation("fromer", "t2", fromTopic, toTopic, (List<Kmessage> messageList, Object kafkaConsumer, AtomicBoolean isNoStop)->{
					for(Kmessage message : messageList)
					{
						//讲一个例子，比如获取订单，修改订单状态，然后，发到库存 topic
					
						//------------------开始数据库事务--------------------------
						//修改状态，消费数据并更换数据
						log.info("开始处理数据zzzzzzzzzz");
						//TODO 处理数据，数据库操作
						log.info("开始填充数据，向下一个topic发送数据");
						//TODP 填充数据
						message.setVal("新数据");
						//-----------------提交或回滚事务----------------------------
						//注意!!!!!!!:出错后，一定要抛出来错误
					}
					 //isNoStop.set(false); 停止topic间数据传送
				});
			} catch (Exception e) {
				System.out.println(e);
				return;
			}
		}).start();
	 */
	public void topicTransformation(String groupId,String tranactionId,Ktopic fromTopic,Ktopic toTopic,CallbackReceiveMessage transformationCallBack);
	
}
